package com.atguigu.gmall.realtime.app.dws;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.bean.ProvinceStats;
import com.atguigu.gmall.realtime.utils.ClickHouseUtil;
import com.atguigu.gmall.realtime.utils.MyKafka;
import com.atguigu.gmall.realtime.utils.MyKafkaPro;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class ProvinceStatsSqlApp {
    public static void main(String[] args) throws Exception {
        //TODO 1.基本环境准备
        //1.1 流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2 设置并行度
        env.setParallelism(4);
        //1.3 表执行环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //TODO 2.从Kafka中读取数据，并转换为动态表
        String groupId = "province_stats";
        String orderWideTopic = "dwm_order_wide";

       /* FlinkKafkaConsumer flinkKafkaConsumer = MyKafka.getFlinkKafkaConsumer(orderWideTopic, groupId);
        DataStreamSource dataStreamSource = env.addSource(flinkKafkaConsumer);
        dataStreamSource.print("dwm宽表流");*/

        tableEnv.executeSql("CREATE TABLE ORDER_WIDE (" +
                "province_id BIGINT, " +
                "province_name STRING," +
                "province_area_code STRING," +
                "province_iso_code STRING," +
                "province_3166_2_code STRING," +
                "order_id STRING, " +
                "split_total_amount DOUBLE," +
                "create_time STRING," +
                "rowtime AS TO_TIMESTAMP(create_time) ," +
                "WATERMARK FOR  rowtime  AS rowtime - INTERVAL '3' SECOND)" +
                " WITH (" +  MyKafkaPro.getKafkaDDL(orderWideTopic,groupId) + ")");

        //TODO 3.分组、开窗、聚合
        Table provinceStateTable = tableEnv.sqlQuery("select " +
                "DATE_FORMAT(TUMBLE_START(rowtime, INTERVAL '10' SECOND ),'yyyy-MM-dd HH:mm:ss') stt, " +
                "DATE_FORMAT(TUMBLE_END(rowtime, INTERVAL '10' SECOND ),'yyyy-MM-dd HH:mm:ss') edt , " +
                "province_id," +
                "province_name," +
                "province_area_code area_code," +
                "province_iso_code iso_code ," +
                "province_3166_2_code iso_3166_2 ," +
                "COUNT( DISTINCT  order_id) order_count, " +
                "sum(split_total_amount) order_amount," +
                "UNIX_TIMESTAMP()*1000 ts "+
                " from  ORDER_WIDE group by  TUMBLE(rowtime, INTERVAL '10' SECOND )," +
                " province_id,province_name,province_area_code,province_iso_code,province_3166_2_code ");



        //TODO 4.转换为数据流
        DataStream<ProvinceStats> provinceStatsDataStream =
                tableEnv.toAppendStream(provinceStateTable, ProvinceStats.class);

        provinceStatsDataStream.print(">>>>");
        //TODO 5.写入到clickHouse
        provinceStatsDataStream.addSink(ClickHouseUtil.
                <ProvinceStats>getJdbcSink("insert into  province_stats_1021  values(?,?,?,?,?,?,?,?,?,?)"));

        env.execute();
    }
}
